package com.lagou.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.HBaseAdmin;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.List;

public class LoadProcessor {

    public static final String PROCESS_JAR = "/processor/process.jar";

    private final Connection connection;
    private final TableName tableName;

    public LoadProcessor(Connection connection, TableName tableName) {
        this.connection = connection;
        this.tableName = tableName;
    }

    public void loadProcess() {
        try {
            uploadJar();
            loadProcess(this.connection, this.tableName);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void uploadJar() throws URISyntaxException, IOException, InterruptedException {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", "hdfs://centos1:8020");
        FileSystem fs = FileSystem.get(new URI("hdfs://centos1:8020"), configuration, "root");
        Path jarPath = new Path(PROCESS_JAR);
        if (fs.exists(jarPath)) {
            fs.delete(jarPath, true);
        }
        fs.mkdirs(new Path("/processor"));
        fs.copyFromLocalFile(new Path("target/zk-1.0-SNAPSHOT.jar"), jarPath);
        boolean exists = fs.exists(jarPath);
        if (exists) {
            System.out.printf("上传 process jar ok %s%n", PROCESS_JAR);
        } else {
            System.out.printf("上传 process jar fail %s%n", PROCESS_JAR);
        }
    }

    private void loadProcess(Connection connection, TableName tableName) throws IOException {
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        HTableDescriptor tableDescriptor = admin.getTableDescriptor(tableName);
        List<String> coprocessors = tableDescriptor.getCoprocessors();
        System.out.println("未挂载协处理器 " + coprocessors);
        admin.disableTable(tableName);
        tableDescriptor.addCoprocessor("com.lagou.hbase.FriendsDeleteProcessor", new Path(PROCESS_JAR), Coprocessor.PRIORITY_USER, null);
        admin.modifyTable(tableName, tableDescriptor);
        admin.enableTable(tableName);
        List<String> coprocessors1 = admin.getTableDescriptor(tableName).getCoprocessors();
        System.out.println("挂载协处理器成功" + coprocessors1);
    }
}
